package com.duowan.realtime.scheduled.batch.listener;

import com.duowan.warn.webservice.client.WarnWebService;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;

import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;


/**
 * 报警listener如果一个job执行失败，发送短信
 * 如果执行时间超过了alterThresholdSecond（默认30秒），发送短信
 * 每次发送短信的间隔由errorMsgDelayMinute(默认30分钟)，指定。
 *
 * @author tangsicheng
 * @version 1.0
 * @since 1.0
 */
public class JobListener implements JobExecutionListener {

    private static final Logger logger = LoggerFactory.getLogger(JobListener.class);

    private static Long DEFAULT_ERROR_MSG_DELAY_MINUTE = 30l;

    private static Long DEFAULT_JOB_TIME_THRESHOLD_SECOND = 60l;

    private static Long errorMsgDelayMinute = DEFAULT_ERROR_MSG_DELAY_MINUTE;

    private static AtomicLong lastSendTime = new AtomicLong();

    private List<String> passPorts = Lists.newLinkedList();

    private Long jobTimeThreshold = DEFAULT_JOB_TIME_THRESHOLD_SECOND;

    private WarnWebService warnWebService = WarnWebService.getInstatnce();


    @Override
    public void beforeJob(JobExecution jobExecution) {

    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        try {
            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
                Long costSecond = getJobCostTime(jobExecution);
                Long dif = costSecond - jobTimeThreshold;
                logger.info(jobExecution.getJobInstance().getJobName() + "执行时间" + costSecond + "秒");
                if (dif > 0) sendAlterMsg(jobExecution.getJobInstance().getJobName() + ",执行超时，超时" + dif + "秒");
            } else {
                sendAlterMsg(jobExecution.getJobInstance().getJobName() + ",执行失败，状态:" + jobExecution.getStatus());
            }
        } catch (Throwable e) {
            logger.error(e.getMessage(), e);
            warnWebService.sendMsg(passPorts, "报警模块失败,msg:" + e.getMessage());
        }

    }

    private Long getJobCostTime(JobExecution jobExecution) {
        Date startTime = jobExecution.getStartTime();
        Date endTime = jobExecution.getEndTime();
        return getTimeDif(startTime.getTime(), endTime.getTime(), TimeUnit.SECONDS);
    }

    private void sendAlterMsg(String s) {
        if (shouldSend()) {
            logger.error(s);
            warnWebService.sendMsg(passPorts, s);
            lastSendTime.set(System.currentTimeMillis());
        }
    }


    private boolean shouldSend() {
        if (lastSendTime == null) return true;
        Long timeFromLastSend = getTimeDif(lastSendTime.get(), System.currentTimeMillis(), TimeUnit.MINUTES);
        return timeFromLastSend >= errorMsgDelayMinute;
    }

    public void setJobTimeThreshold(Long jobTimeThreshold) {
        this.jobTimeThreshold = jobTimeThreshold;
    }

    public void setPassPorts(List<String> passPorts) {
        this.passPorts = passPorts;
    }

    private long getTimeDif(Long startTime, Long endTime, TimeUnit timeUnit) {
        return timeUnit.convert(endTime - startTime, TimeUnit.MILLISECONDS);
    }


    public void setErrorMsgDelayMinute(Long errorMsgDelayMinute) {
        this.errorMsgDelayMinute = errorMsgDelayMinute;
    }
}
